#ifndef __M_BROKER_H__
#define __M_BROKER_H__

#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"
#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_proto.pb.h"
#include "../mqcommon/mq_threadPool.hpp"
#include "mq_vhost.hpp"
#include "mq_connection.hpp"
#include "mq_vhost.hpp"
#include "mq_consumer.hpp"
#include <iostream>
#include <unordered_map>
#include <functional>
namespace wsmq
{
#define HOSTNAME "MyVirtualHost"
#define DBFILE "/meta.db"
    class Server
    {
    public:
        typedef std::shared_ptr<google::protobuf::Message> MessagePtr;

        Server(int port, const std::string &basedir) : _server(&_baseloop,
                                                               muduo::net::InetAddress("0.0.0.0", port),
                                                               "Server", muduo::net::TcpServer::kReusePort),
                                                       _dispatcher(std::bind(&Server::OnUnknownMessage, this,
                                                                             std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)),
                                                       _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher,
                                                                                                        std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
                                                       _virtual_host(std::make_shared<VirtualHost>(HOSTNAME, basedir, basedir + DBFILE)),
                                                       _consumer_manager(std::make_shared<ConsumerManager>()),
                                                       _connection_manager(std::make_shared<ConnectionManager>()),
                                                       _threadpool(std::make_shared<ThreadPool>())
        {
            // 用恢复的队列历史消息初始化消费者管理结构
            MsgQueueMap qm = _virtual_host->allQueues();
            for (auto it : qm)
            {
                _consumer_manager->initQueueConsumer(it.first);
            }

            // 注册业务请求处理函数
            _dispatcher.registerMessageCallback<wsmq::openChannelRequest>(std::bind(&Server::onOpenChannel, this,
                                                                                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<wsmq::closeChannelRequest>(std::bind(&Server::onCloseChannel, this,
                                                                                     std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<wsmq::declareExchangeRequest>(std::bind(&Server::ondeclareExchange, this,
                                                                                        std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<wsmq::removeExchangeRequest>(std::bind(&Server::onremoveExchange, this,
                                                                                       std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<wsmq::declareQueueRequest>(std::bind(&Server::ondeclareQueue, this,
                                                                                     std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<wsmq::removeQueueRequest>(std::bind(&Server::onremoveQueue, this,
                                                                                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<wsmq::queueBindRequest>(std::bind(&Server::onqueueBind, this,
                                                                                  std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<wsmq::QueueUnBindRequest>(std::bind(&Server::onQueueUnBind, this,
                                                                                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<wsmq::basicPublishRequest>(std::bind(&Server::onbasicPublish, this,
                                                                                     std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<wsmq::basicAckRequest>(std::bind(&Server::onbasicAck, this,
                                                                                 std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<wsmq::basicConsumeRequest>(std::bind(&Server::onbasicConsume, this,
                                                                                     std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<wsmq::basicCancelRequest>(std::bind(&Server::onbasicCancel, this,
                                                                                    std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage,
                                                 _codec.get(), std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            // 设置连接回调处理函数
            _server.setConnectionCallback(std::bind(&Server::Onconnection, this, std::placeholders::_1));
        }
        void Start()
        {
            _server.start();
            _baseloop.loop();
        }

    private:
        void onOpenChannel(const muduo::net::TcpConnectionPtr &conn, const openChannelRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("打开信道时，未找到连接对应的连接对象！");
                conn->shutdown();
                return;
            }
            myconn->openChannel(message);
        }
        void onCloseChannel(const muduo::net::TcpConnectionPtr &conn, const closeChannelRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("关闭信道时，未找到连接对应的连接对象！");
                conn->shutdown();
                return;
            }
            myconn->closeChannel(message);
        }
        void ondeclareExchange(const muduo::net::TcpConnectionPtr &conn, const declareExchangeRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("声明交换机时，未找到连接对应的连接对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr chp = myconn->getChannel(message->cid());
            if (chp.get() == nullptr)
            {
                DLOG("声明交换机时，没有找到信道！");
                return;
            }
            chp->declareExchange(message);
        }
        void onremoveExchange(const muduo::net::TcpConnectionPtr &conn, const removeExchangeRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("移除交换机时，未找到连接对应的连接对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr chp = myconn->getChannel(message->cid());
            if (chp.get() == nullptr)
            {
                DLOG("移除交换机时，没有找到信道！");
                return;
            }
            chp->removeExchange(message);
        }
        void ondeclareQueue(const muduo::net::TcpConnectionPtr &conn, const declareQueueRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("声明队列时，未找到连接对应的连接对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr chp = myconn->getChannel(message->cid());
            if (chp.get() == nullptr)
            {
                DLOG("声明队列时，没有找到信道！");
                return;
            }
            chp->declareQueue(message);
        }
        void onremoveQueue(const muduo::net::TcpConnectionPtr &conn, const removeQueueRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("移除队列时，未找到连接对应的连接对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr chp = myconn->getChannel(message->cid());
            if (chp.get() == nullptr)
            {
                DLOG("移除队列时，没有找到信道！");
                return;
            }
            chp->removeQueue(message);
        }
        void onqueueBind(const muduo::net::TcpConnectionPtr &conn, const queueBindRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("绑定队列时，未找到连接对应的连接对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr chp = myconn->getChannel(message->cid());
            if (chp.get() == nullptr)
            {
                DLOG("绑定队列时，没有找到信道！");
                return;
            }
            chp->queueBind(message);
        }
        void onQueueUnBind(const muduo::net::TcpConnectionPtr &conn, const QueueUnBindRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("解绑队列时，未找到连接对应的连接对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr chp = myconn->getChannel(message->cid());
            if (chp.get() == nullptr)
            {
                DLOG("解绑队列时，没有找到信道！");
                return;
            }
            chp->QueueUnBind(message);
        }
        void onbasicPublish(const muduo::net::TcpConnectionPtr &conn, const basicPublishRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("订阅消息时，未找到连接对应的连接对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr chp = myconn->getChannel(message->cid());
            if (chp.get() == nullptr)
            {
                DLOG("订阅消息时时，没有找到信道！");
                return;
            }
            chp->basicPublish(message);
        }
        void onbasicConsume(const muduo::net::TcpConnectionPtr &conn, const basicConsumeRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("队列消息订阅时，未找到连接对应的连接对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr chp = myconn->getChannel(message->cid());
            if (chp.get() == nullptr)
            {
                DLOG("队列消息订阅时，没有找到信道！");
                return;
            }
            chp->basicConsume(message);
        }
        void onbasicCancel(const muduo::net::TcpConnectionPtr &conn, const basicCancelRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("队列消息取消订阅时，未找到连接对应的连接对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr chp = myconn->getChannel(message->cid());
            if (chp.get() == nullptr)
            {
                DLOG("队列消息取消订阅时，没有找到信道！");
                return;
            }
            chp->basicCancel(message);
        }
        void onbasicAck(const muduo::net::TcpConnectionPtr &conn, const basicAckRequestPtr &message, muduo::Timestamp)
        {
            Connection::ptr myconn = _connection_manager->getConnection(conn);
            if (myconn.get() == nullptr)
            {
                DLOG("确认消息时，未找到连接对应的连接对象！");
                conn->shutdown();
                return;
            }
            Channel::ptr chp = myconn->getChannel(message->cid());
            if (chp.get() == nullptr)
            {
                DLOG("确认消息时，没有找到信道！");
                return;
            }
            chp->basicAck(message);
        }

        void Onconnection(const ::muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                // 根据当前连接，实例化连接对象
                _connection_manager->createConnection(conn, _virtual_host, _consumer_manager, _codec, _threadpool);
            }
            else
            {
                // 销毁连接对象
                _connection_manager->deleteConnection(conn);
            }
        }
        void OnUnknownMessage(const muduo::net::TcpConnectionPtr &conn, const MessagePtr &message, muduo::Timestamp)
        {
            LOG_INFO << "On Unknown Message" << message->GetTypeName();
            conn->shutdown();
        }

    private:
        muduo::net::EventLoop _baseloop;
        muduo::net::TcpServer _server;  // 服务器对象
        ProtobufDispatcher _dispatcher; // 请求分发器对象 ---注册请求处理函数
        ProtobufCodecPtr _codec;        // protobuf协议处理对象 ---针对收到的数据进行protobuf协议处理
        VirtualHost::ptr _virtual_host;
        ConsumerManager::ptr _consumer_manager;
        ConnectionManager::ptr _connection_manager;
        ThreadPool::ptr _threadpool;
    };

}

#endif